package org.nbict.iot.task.log;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.kafka.spout.KafkaSpoutConfig;
import org.apache.storm.kafka.spout.trident.KafkaTridentSpoutOpaque;
import org.apache.storm.trident.Stream;
import org.apache.storm.trident.TridentTopology;
import org.apache.storm.tuple.Fields;
import org.nbict.iot.PropertyUtil;
import org.nbict.iot.task.realtime.func.PacketJsonProject;
import org.nbict.iot.task.realtime.func.PacketResolver;
import org.nbict.iot.trident.kafka.KafkaSpout4ParsedTools;
import org.nbict.iot.trident.mogo.mapper.MongoMapper;
import org.nbict.iot.trident.mogo.mapper.impl.SimpleMongoMapper;
import org.nbict.iot.trident.mogo.state.MongoState;
import org.nbict.iot.trident.mogo.state.MongoStateFactory;
import org.nbict.iot.trident.mogo.state.MongoStateUpdater;

/**
 * 日志记录，需要增加插入时间
 * Created by songseven on 18/6/27.
 */
public class LogTrident {

    
    public static void main(String[] args) throws InvalidTopologyException, AuthorizationException, AlreadyAliveException {

        //设置kafka消费策略
        KafkaSpoutConfig.FirstPollOffsetStrategy strategy = KafkaSpoutConfig
                .FirstPollOffsetStrategy.EARLIEST;
        if (args.length > 1) {
            strategy = KafkaSpoutConfig.FirstPollOffsetStrategy.valueOf
                    (args[1]); //strategy name
        }

        String url = PropertyUtil.getProperty("mongo.url");
        String database = PropertyUtil.getProperty("mongo.database");
        String username = PropertyUtil.getProperty("mongo.username");
        String password = PropertyUtil.getProperty("mongo.password");
        String admin = PropertyUtil.getProperty("mongo.admin.name");
        String collectionName = "car_history";

        TridentTopology topology = new TridentTopology();

        KafkaSpoutConfig spoutConfig = KafkaSpout4ParsedTools.newKafkaSpoutConfig
                ("192.168.1.83:9092", "transit-.*-simple",  //192.168.1.83:9092
                        "iot-car-group-log", strategy); //group id ignored
        KafkaTridentSpoutOpaque kafkaSpout = KafkaSpout4ParsedTools.newKafkaTridentSpoutOpaque(spoutConfig);

        //重要，由流名称决定消费者组
        Stream stream = topology.newStream("iot-car-parsed-4-log", kafkaSpout);

        Fields fields = new Fields("platform_id", "car_vin","action", "timestamp", "upload");

        MongoMapper mapper = new SimpleMongoMapper().withFields("platform_id", "car_vin","action", "timestamp", "upload");

        MongoState.Options options = new MongoState.Options()
                .withUrl(url)
                .withDatabase(database)
                .withUsername(username)
                .withPassword(password)
                .withAdmin(admin)
                .withCollectionName(collectionName)
                .withMapper(mapper);

        Fields packetFields = new Fields("car_packet_header", "car_packet_body");
        Fields outputFields = new Fields("car_vin", "action", "timestamp", "upload");
        Stream stage = stream.each(new Fields("platform_id", "packet"), new PacketJsonProject(packetFields), packetFields)
                .each(new Fields("platform_id", "car_packet_header", "car_packet_body"), new PacketResolver(), outputFields)
                .project(fields); //去除packet元组

        stage.partitionPersist( new MongoStateFactory(options), fields,
                new MongoStateUpdater(), new Fields());

        Config config = new Config();
        config.put("kafka.topic.wildcard.match", true);

        if (args.length > 0) {
            //集群提交模式
            config.setDebug(false);
            StormSubmitter.submitTopology(args[0], config, topology.build());
        } else {
            //本地测试模式
            config.setDebug(false);
            //设置1个进程
            config.setNumWorkers(1);
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("iot-car-realtime", config, topology.build());
        }
    }
}
